package com.daffodil.workable.schedule;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Date;
import java.util.List;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;

import javax.sql.DataSource;

import com.daffodil.core.entity.Query;
import com.daffodil.util.StringUtils;
import com.daffodil.workable.constant.WorkableConstant;
import com.daffodil.workable.entity.WorkablePlan;
import com.daffodil.workable.util.WorkableUtils;

import lombok.extern.slf4j.Slf4j;

/**
 * -定时轮询任务
 * @author yweijian
 * @date 2022年4月27日
 * @version 1.0
 * @description
 */
@Slf4j
public class WorkableScheduleThread {

    private DataSource dataSource;
    private Integer blockingQueueSize;
    private Thread planThread;
    private Thread jobThread;

    private volatile boolean planStopStatus = false;
    private volatile boolean jobStopStatus = false;
    private volatile static LinkedBlockingDeque<WorkablePlan> jobDatas;

    private static final long ONE_SECONDS = 1000L;
    private static final long FIVE_SECONDS = ONE_SECONDS * 5L;

    private static WorkableScheduleThread instance = new WorkableScheduleThread();

    public static WorkableScheduleThread getInstance() {
        return instance;
    }

    public void init(DataSource dataSource, int blockingQueueSize) {
        this.dataSource = dataSource;
        this.blockingQueueSize = blockingQueueSize;
        jobDatas = new LinkedBlockingDeque<WorkablePlan>(blockingQueueSize);
    }

    public void start() {
        this.startPlanThread();
        this.startJobThread();
    }

    /**
     * -plan计划每5秒执行一次
     */
    private void startPlanThread() {
        this.planThread = new Thread(() -> {
            while (!this.planStopStatus) {
                try {
                    //时间误差纠整
                    TimeUnit.MILLISECONDS.sleep(FIVE_SECONDS - System.currentTimeMillis() % ONE_SECONDS);
                } catch (InterruptedException e) {
                    log.error(e.getMessage(), e);
                }
                if(this.dataSource != null) {
                    Connection connection = null;
                    Boolean autoCommit = null;
                    PreparedStatement prepared = null;
                    try {
                        connection = dataSource.getConnection();
                        autoCommit = connection.getAutoCommit();
                        connection.setAutoCommit(false);
                        prepared = connection.prepareStatement("select * from wok_lock where id='wok_lock' for update");
                        boolean lock = prepared.execute();
                        if(lock) {
                            Query<WorkablePlan> query = new Query<WorkablePlan>();
                            WorkablePlan workablePlan = new WorkablePlan();
                            workablePlan.setStatus(WorkableConstant.WORKABLE_STARTUP);
                            long currentTime = System.currentTimeMillis();
                            long nextTime = FIVE_SECONDS + currentTime;
                            workablePlan.setNextTime(nextTime);
                            query.setEntity(workablePlan);
                            List<WorkablePlan> plans = WorkableUtils.getPlanService().selectWorkablePlanList(query);
                            
                            if(StringUtils.isNotEmpty(plans)) {
                                for(int i = 0; i < plans.size(); i++) {
                                    WorkablePlan plan = plans.get(i);
                                    long planTime = plan.getNextTime();
                                    if(currentTime <= planTime && planTime <= nextTime) {
                                        if(jobDatas.size() <= this.blockingQueueSize) {
                                            jobDatas.add(plan);
                                        }else {
                                            if(log.isWarnEnabled()) {
                                                log.warn("定时执行可拖拽编排任务计划线程池已满，执行计划任务丢弃planId={}，任务taskId={}", plan.getId(), plan.getTaskId());
                                            }
                                        }
                                    }
                                    long nextTriggerTime = WorkableUtils.getNextTriggerTime(plan.getExpression(), new Date(nextTime));
                                    plan.setLastTime(plan.getNextTime());
                                    plan.setNextTime(nextTriggerTime);
                                    WorkableUtils.getPlanService().updateWorkablePlan(plan);
                                }
                            }
                        }
                    }catch (Exception e) {
                        log.error(e.getMessage(), e);
                    }finally {
                        if (connection != null) {
                            try {
                                connection.commit();
                                connection.setAutoCommit(autoCommit);
                                connection.close();
                            } catch (SQLException e) {
                                log.error(e.getMessage(), e);
                            }
                        }
                        if (prepared != null) {
                            try {
                                prepared.close();
                            } catch (SQLException e) {
                                log.error(e.getMessage(), e);
                            }
                        }
                    }
                }
            }
        });
        planThread.setDaemon(true);
        planThread.start();
    }

    /**
     * -job任务每秒执行一次
     */
    private void startJobThread() {
        jobThread = new Thread(() -> {
            while (!jobStopStatus) {
                WorkablePlan plan = jobDatas.poll();
                try {
                    //时间误差纠整
                    TimeUnit.MILLISECONDS.sleep(ONE_SECONDS - System.currentTimeMillis() % ONE_SECONDS);
                } catch (InterruptedException e) {
                    log.error(e.getMessage(), e);
                }
                WorkableTriggerThread.add(plan);
            }
        });
        jobThread.setDaemon(true);
        jobThread.start();
    }

    public void stop() {
        this.stopPlanThread();
        this.stopJobThread();
    }

    private void stopPlanThread() {
        this.planStopStatus = true;
        try {
            TimeUnit.SECONDS.sleep(1L);
        } catch (InterruptedException e) {
            log.error(e.getMessage(), e);
        }
        if (planThread != null && planThread.getState() != Thread.State.TERMINATED){
            planThread.interrupt();
            try {
                planThread.join();
            } catch (InterruptedException e) {
                log.error(e.getMessage(), e);
            }
        }
    }

    private void stopJobThread() {
        this.jobStopStatus = true;
        try {
            TimeUnit.SECONDS.sleep(1L);
        } catch (InterruptedException e) {
            log.error(e.getMessage(), e);
        }
        if (jobThread != null && jobThread.getState() != Thread.State.TERMINATED){
            jobThread.interrupt();
            try {
                jobThread.join();
            } catch (InterruptedException e) {
                log.error(e.getMessage(), e);
            }
        }
    }

}
